package com.zheng.umbrella.web.jms;

import com.zheng.common.constant.MqConstant;
import com.zheng.common.util.DateUtil;
import com.zheng.common.util.RedisUtil;
import com.zheng.common.util.StringUtil;
import com.zheng.umbrella.common.constant.UmbrellaConstant;
import com.zheng.umbrella.dao.model.UmbrellaHardwareMachine;
import com.zheng.umbrella.dao.model.UmbrellaHardwareMachineExample;
import com.zheng.umbrella.rpc.api.UmbrellaHardwareMachineService;
import com.zheng.upms.rpc.api.UpmsSystemParamService;
import org.apache.activemq.advisory.AdvisorySupport;
import org.apache.activemq.command.*;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import javax.jms.Message;
import javax.jms.MessageListener;

/**
 * MQ消费者
 * Created by shuzheng on 2017/2/19.
 */
public class DefaultMessageQueueListener implements MessageListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(DefaultMessageQueueListener.class);

    @Autowired
    ThreadPoolTaskExecutor threadPoolTaskExecutor;
    @Autowired
    private UmbrellaHardwareMachineService umbrellaHardwareMachineService;
    @Autowired
    private UpmsSystemParamService upmsSystemParamService;

    private String cacheModule = "zheng-umbrella-web";
    /**
     *  超时门限(秒)
     */
    private static Integer EXPIRE_TIME = 60 * 60 * 24 * 100;

    @Override
    public void onMessage(final Message message) {
        // 使用线程池多线程处理
        threadPoolTaskExecutor.execute(new Runnable() {
            @Override
            public void run() {
                //JMS规范中的消息类型包括TextMessage、MapMessage、ObjectMessage、BytesMessage、和StreamMessage等五种
                if(message instanceof ActiveMQMessage){
                    ActiveMQMessage mqMessage=(ActiveMQMessage)message;
                    if(mqMessage.getDataStructure() instanceof ConnectionInfo){
                        //destination = topic://ActiveMQ.Advisory.Connection
                        ConnectionInfo connectionInfo=(ConnectionInfo)mqMessage.getDataStructure();
                        String clientId = connectionInfo.getClientId();
                        //String clentIp = connectionInfo.getClientIp();
                        String connectionId = connectionInfo.getConnectionId().getValue();
                        String seqNum = connectionId.substring(connectionId.lastIndexOf(":")+1,connectionId.length());
                        connectionId = connectionId.replaceAll(":","");
                        String skeyPath = cacheModule + ":" + UmbrellaConstant.UM_UMBRELLA_MACHINE_STATUS + ":" + connectionId;
                        if(clientId!=null && (clientId.startsWith("bjsv") || clientId.startsWith("test") || clientId.startsWith("dev") || clientId.indexOf("SJ")>-1)){
                            String keypath = RedisUtil.get(clientId+"_keyPath");
                            if(StringUtils.isNotBlank(keypath)){
                                RedisUtil.remove(keypath);
                            }
                            //放进缓存
                            RedisUtil.set(clientId, seqNum);
                            RedisUtil.set(skeyPath, clientId);
                            RedisUtil.set(clientId+"_keyPath", skeyPath);

                            UmbrellaHardwareMachineExample example = new UmbrellaHardwareMachineExample();
                            UmbrellaHardwareMachineExample.Criteria criteria = example.createCriteria();
                            criteria.andCodeEqualTo(clientId);
                            criteria.andStatusEqualTo(UmbrellaConstant.UM_MACHINE_STATUS_3);
                            UmbrellaHardwareMachine machine = umbrellaHardwareMachineService.selectFirstByExample(example);
                            if(machine!=null){
                                //更新状态为在线
                                UmbrellaHardwareMachine newMachine = new UmbrellaHardwareMachine();
                                newMachine.setStatus(UmbrellaConstant.UM_MACHINE_STATUS_2);
                                umbrellaHardwareMachineService.updateByExampleSelective(newMachine,example);
                            }

                        }
                        //更新伞机状态
                    }else if(mqMessage.getDataStructure() instanceof RemoveInfo){
                        //topic://ActiveMQ.Advisory.Connection
                        //topic://ActiveMQ.Advisory.Consumer.Topic.bjs.server.topic.umbrella.machine
                        RemoveInfo removeInfo=(RemoveInfo)mqMessage.getDataStructure();

                        String objectId = removeInfo.getObjectId().toString();
                        String seqNum = objectId.substring(objectId.lastIndexOf(":")+1,objectId.length());
                        objectId = objectId.replaceAll(":","");
                        String skeyPath = cacheModule + ":" + UmbrellaConstant.UM_UMBRELLA_MACHINE_STATUS + ":" + objectId;
                        String sValue = RedisUtil.get(skeyPath);

                        //删除缓存
                        RedisUtil.remove(skeyPath);

                        if(StringUtils.isNotEmpty(sValue)){
                            String preSeqNum = RedisUtil.get(sValue);
                            //删除缓存
                            RedisUtil.remove(sValue);
                            if(StringUtils.isNotBlank(preSeqNum) && !seqNum.equals(preSeqNum))return;

                            UmbrellaHardwareMachineExample example = new UmbrellaHardwareMachineExample();
                            UmbrellaHardwareMachineExample.Criteria criteria = example.createCriteria();
                            criteria.andCodeEqualTo(sValue);
                            criteria.andStatusEqualTo(UmbrellaConstant.UM_MACHINE_STATUS_2);
                            UmbrellaHardwareMachine machine = umbrellaHardwareMachineService.selectFirstByExample(example);

                            if(machine!=null){
                                //更新状态为离线
                                UmbrellaHardwareMachine newMachine = new UmbrellaHardwareMachine();
                                newMachine.setStatus(UmbrellaConstant.UM_MACHINE_STATUS_3);
                                umbrellaHardwareMachineService.updateByExampleSelective(newMachine,example);
                            }
                        }
                        //if(mqMessage.getDestination().getPhysicalName().equals(AdvisorySupport.CONNECTION_ADVISORY_TOPIC.getTopicName())){
                        //    System.out.println(mqMessage.getDestination().getPhysicalName().substring(AdvisorySupport.CONSUMER_ADVISORY_TOPIC_PREFIX.length(),mqMessage.getDestination().getPhysicalName().length())+"掉了一个客户端");
                        //}
                        //消费者暂无需订阅，如需订阅还需修改applicationContext-activemq.xml配置
                        /*
                        else if(mqMessage.getDestination().getPhysicalName().startsWith(AdvisorySupport.CONSUMER_ADVISORY_TOPIC_PREFIX)){
                            System.out.println(mqMessage.getDestination().getPhysicalName().substring(AdvisorySupport.CONSUMER_ADVISORY_TOPIC_PREFIX.length(),mqMessage.getDestination().getPhysicalName().length())+"掉了一个客户端");
                        }*/
                    }
                    //消费者暂无需订阅，如需订阅还需修改applicationContext-activemq.xml配置
                    /*else if(mqMessage.getDataStructure() instanceof ConsumerInfo){
                        //topic://ActiveMQ.Advisory.Consumer.Topic.bjs.server.topic.umbrella.machine
                        ConsumerInfo consumerInfo=(ConsumerInfo)mqMessage.getDataStructure();
                        System.out.println((consumerInfo.getDestination().getPhysicalName())+"上来了一个客户端："+consumerInfo.getSubscriptionName());
                    }*/
                }
            }
        });
    }

}
